package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo11Burk {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", "2")
      .master("local")
      .appName("sql")
      .getOrCreate()

    val df: DataFrame = spark
      .read
      .option("sep", ",")
      .schema("burk STRING , year STRING , tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE,tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE,tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
      .csv("spark/data/burks.txt")

    df.createOrReplaceTempView("burk")


    spark
      .sql("select *,array(tsl01,tsl02,tsl03,tsl04,tsl05,tsl06,tsl07,tsl08,tsl09,tsl10,tsl11,tsl12) as a from burk")
    //.show(truncate = false)

    spark.sql(
      """
        |
        |
        |select burk,year,month,income ,sum(income) over(partition by burk,year order by month rows unbounded preceding ) as sum_income
        |from
        |(
        |select burk,year,month,income from
        |burk
        |LATERAL VIEW explode(map('01',tsl01,'02',tsl02,'03',tsl03,'04',tsl04,'05',tsl05,'06',tsl06,'07',tsl07,'08',tsl08,'09',tsl09,'10',tsl10,'11',tsl11,'12',tsl12)) a as month,income
        |) as a
        |
        |
      """.stripMargin) //.show(100)


    /**
      *
      * 1、统计每个公司每年按月累计收入
      *
      * 输出结果
      * 公司代码,年度,月份,当月收入,累计收入
      *
      */

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val m: Column = map(
      expr("01"), $"tsl01",
      expr("02"), $"tsl02",
      expr("03"), $"tsl03",
      expr("04"), $"tsl04",
      expr("05"), $"tsl05",
      expr("06"), $"tsl06",
      expr("07"), $"tsl07",
      expr("08"), $"tsl08",
      expr("09"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )


    df.select($"burk", $"year", explode(m).as(Array("month", "income")))
      .select($"burk", $"year", $"month", $"income", sum($"income") over Window.partitionBy($"burk", $"year").orderBy($"month").rowsBetween(Window.unboundedPreceding, Window.currentRow) as "sum_income")
    //.show()


    /**
      *
      * 2、统计每个公司当月比上年同期增长率
      * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
      *
      */

    df.select($"burk", $"year", explode(m).as(Array("month", "income"))) //行列转换
      .select($"burk", $"year", $"month", $"income", lag($"income", 1, 0.0) over Window.partitionBy($"burk", $"month").orderBy($"year") as "last_income")
      .select($"burk", $"year", $"month", $"income", $"income" / $"last_income" - 1 as "rate")
      .select($"burk", $"year", $"month", $"income", when($"rate".isNull, 1.0).otherwise($"rate") as "rate")
      .show(100)
  }
}
